package com.eeboot;

import com.eeboot.vo.Input;
import com.eeboot.vo.Output;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class WindowTest {
    public static void main(String[] args) throws Exception{
        //参数检查
        if (args.length != 2) {
            System.err.println("USAGE:\nWindowTest <hostname> <port>");
            return;
        }

        String hostname = args[0];
        Integer port = Integer.parseInt(args[1]);


        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        //获取数据
        DataStreamSource<String> stream = env.socketTextStream(hostname, port);
        stream.map((MapFunction<String, Input>) value -> {
            String [] arr = value.split(",");
            return Input.of(arr[0], arr[1], Long.valueOf(arr[2]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Input>(Time.seconds(1)) {
            @Override
            public long extractTimestamp(Input element) {
                return element.millTime;
            }
        }).keyBy("profile")
                .timeWindow(Time.minutes(1), Time.minutes(1))
                .trigger(DefTrigger.create())
                .apply(new WindowFunction<Input, Output, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<Input> input, Collector<Output> out) throws Exception {
                        String profile = null;
                        Integer count = 0;
                        for(Input i: input){
                            profile = i.profile;
                            count += 1;
                        }
                        out.collect(Output.of(profile, count));
                    }
                }).print();

        env.execute("window test");
    }
}
